package com.example.controller.kafka;

import com.fasterxml.jackson.databind.JsonNode;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.connect.json.JsonDeserializer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

/**
 * @program: kafka-study
 * @description:
 * @author: yyyyjinying
 * @create: 2023-10-28 18:36
 **/
public class Consumer {
    public static final String brokerList = "node1:9092,node2:9092,node3:9092";
    public static final String topic = "fullfillment.mydb.user_info";
    public static final String groupId = "first-group";


    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
//        properties.put("value.deserializer", "org.apache.kafka.connect.json.JsonDeserializer"); // 使用 JSON 反序列化器
        properties.put("value.deserializer", JsonDeserializer.class.getName()); // 使用 JSON 反序列化器

        properties.put("bootstrap.servers", brokerList);
        properties.put("group.id", groupId);

        KafkaConsumer<String, JsonNode> consumer = new KafkaConsumer<>(properties); // 使用 JsonNode 来接收消息值
        consumer.subscribe(Collections.singletonList(topic));

        while (true) {
            ConsumerRecords<String, JsonNode> records = consumer.poll(Duration.ofMillis(10000));
            for (ConsumerRecord<String, JsonNode> record : records) {
                JsonNode value = record.value();
                if (value != null) {
                    value = value.get("payload");
                }

                String operation = value.get("op").asText();

                if ("c".equals(operation)) { // 插入
                    JsonNode after = value.get("after");
                    System.out.println("Inserted record: " + after);
                } else if ("u".equals(operation)) { // 更新
                    JsonNode before = value.get("before");
                    JsonNode after = value.get("after");
                    System.out.println("Updated from: " + before + " to: " + after);
                } else if ("d".equals(operation)) { // 删除
                    JsonNode before = value.get("before");
                    System.out.println("Deleted record: " + before);
                }
            }
        }

    }
}
